package DianShang_2024.ds_03.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, current_timestamp, date_format, desc, lit, min, row_number, to_timestamp}

import java.text.SimpleDateFormat
import java.util.Date

object clean04 {
  def main(args: Array[String]): Unit = {
    /*
          4、抽取ods库base_region表中昨天的分区（子任务一生成的分区）数据，并结合dim_region最新分区现有的数据，根据id合并数据到dwd库中
          dim_region的分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以id为合并字段，根据create_time排序取最新的一条），分区字段
          为etl_date且值与ods库的相对应表该值相等，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四
          列,其中dwd_insert_user、dwd_modify_user均填写“user1”。若该条数据第一次进入数仓dwd层则dwd_insert_time、dwd_modify_time均
          填写当前操作时间，并进行数据类型转换。若该数据在进入dwd层时发生了合并修改，则dwd_insert_time时间不变，dwd_modify_time存当前操作时
          间，其余列存最新的值。使用hive cli在表dwd.dim_region最新分区中，查询该分区中数据的条数，将结果内容截图粘贴至客户端桌面【Release\任
          务B提交结果.docx】中对应的任务序号下；
     */

    //  准备环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第四题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd03")

    //  1.使用sql的方式

    //  读取ods层的数据并且对字段的值进行处理
//    spark.sql(
//      """
//        |select
//        |id,
//        |region_name,
//        |create_time,
//        |"user1" as  dwd_insert_user,
//        |to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss")  as dwd_insert_time,
//        |"user1" as dwd_modify_user,
//        |to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss") as dwd_modify_time,
//        |etl_date
//        |from ods03.base_region
//        |where etl_date='20240101'
//        |""".stripMargin).createOrReplaceTempView("ods")
//
//    //  读取dwd层最新分区的数据
//    spark.sql("select * from dwd03.dim_region where etl_date='20240101'")
//      .createOrReplaceTempView("dwd")
//
//    //  将两张表的数据合并，进行开窗
//    val result=spark.sql(
//      """
//        |select
//        |*
//        |from(
//        |select
//        |id,
//        |region_name,
//        |create_time,
//        |dwd_insert_user,
//        |min(dwd_insert_time) over(partition by id order by create_time desc) as dwd_insert_time,
//        |dwd_modify_user,
//        |dwd_modify_time,
//        |row_number() over(partition by id order by create_time desc) as row
//        |from(
//        |select * from dwd
//        |union all
//        |select * from ods
//        |) as t1
//        |) as t2
//        |where row=1
//        |""".stripMargin)
//
//    //  准备插入数据
//    result.drop("row").createOrReplaceTempView("result")
//
//    spark.sql(
//      """
//        |insert into table dwd03.dim_region
//        |partition(etl_date='20240101')
//        |select * from result
//        |""".stripMargin)


    //  2.使用dataframe的方式

    //  读取到ods层的旧分区数据
    val ods=spark.sql("select * from ods03.base_region")
      .where(col("etl_date")==="20240101")
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        lit(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())).cast("timestamp")
      )
      .withColumn("dwd_modify",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )

    //  读取dwd层新分区的数据
    val dwd=spark.sql("select * from dwd03.dim_region")
      .where(col("etl_date")==="20240101")

    //  合并两张表并且开窗函数
    dwd.unionAll(ods)
      .withColumn(
        "dwd_insert_time",
        min("dwd_insert_time") over(Window.partitionBy("id"))
      )
      .withColumn(
        "row",
        row_number() over(Window.partitionBy("id").orderBy(desc("create_time")))
      )
      .where(col("row")===1)
      .drop("row")
      .withColumn("dwd_insert_time",col("dwd_insert_time").cast("timestamp"))
      .withColumn("dwd_modify_time",col("dwd_modify_time").cast("timestamp"))
      .withColumn("etl_date",lit("20240101"))
      .write.mode("append")
      .format("hive")
      .partitionBy("etl_date")
      .saveAsTable("dim_region")





    //  关闭环境
    spark.close()
  }

}
